package com.jython.flink;

import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.SplitStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.lang.reflect.Parameter;
import java.util.ArrayList;

public class WordCount1 {
    public static void main(String[] args) {
        int port;
        try {
            ParameterTool parameterTool = ParameterTool.fromArgs(args);
            port = parameterTool.getInt("port");
        }catch (Exception e){
            System.out.println("没有指定port,默认为9000");
            port=9000;
        }
        StreamExecutionEnvironment env= StreamExecutionEnvironment.getExecutionEnvironment();
        DataStream<String> textStream= env.socketTextStream("192.168.0.180", port, "\n");
        SplitStream<String> split = textStream.split(x -> {
            ArrayList<String> strings = new ArrayList<>();
            Integer integer = Integer.valueOf(x);
            if (integer % 2 == 0) {
                strings.add("even");
            } else {
                strings.add("odd");
            }
            return strings;
        });
        DataStream<String> even = split.select("even").map(x -> {
            System.out.println("这个是偶数:" + x);
            return x;
        });
        DataStream<String> odd= split.select("even").map(x -> {
            System.out.println("这个是奇数:" + x);
            return x;
        });
        DataStream<String> map = split.select("even", "odd").map(x -> {
            System.out.println("这个是综合:" + x);
            return x;
        });
        try {
            env.execute("ok");
        } catch (Exception e) {
            e.printStackTrace();
        }

    }
}
